using System;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using Me.Prettyprint.Cassandra.Utils;
using Me.Prettyprint.Cassandra.Connection.Client;
using Me.Prettyprint.Cassandra.Connection.Factory;
using Me.Prettyprint.Cassandra.Service;
using Me.Prettyprint.Hector.Api.Exceptions;

namespace Me.Prettyprint.Cassandra.Connection
{
    /// <summary>
    /// ConcurrentHClientPool is the default connection pool. Users do not interact
    /// with the pool directly. HConnectionManager is the one responsible for making
    /// calls and managing connection pools.
    /// </summary>
	public class ConcurrentHClientPool : HClientPool
	{
		private static readonly Logger logger = new Logger(typeof(ConcurrentHClientPool));
		
		private ConcurrentQueue<HClient> availableClientQueue;
        private AtomicInt activeClientsCount;
        private AtomicInt realActiveClientsCount;
        private AtomicInt numBlocked;
        private AtomicValue<bool> active;
		private CassandraHost cassandraHost;
		
		private long maxWaitTimeWhenExhausted;
		private HClientFactory clientFactory;
		private CassandraClientMonitor monitor;
		
		public ConcurrentHClientPool(HClientFactory clientFactory, CassandraHost host, CassandraClientMonitor monitor)
		{
			this.clientFactory = clientFactory;
			this.cassandraHost = host;
			this.monitor = monitor;
			
			availableClientQueue = new ConcurrentQueue<HClient>();
			activeClientsCount = new AtomicInt(0);
			realActiveClientsCount = new AtomicInt(0);
			numBlocked = new AtomicInt(0);
			active = new AtomicValue<bool>(true);
			
			maxWaitTimeWhenExhausted = cassandraHost.MaxWaitTimeWhenExhausted < 0?0:cassandraHost.MaxWaitTimeWhenExhausted;
			
			for (int i = 0; i < cassandraHost.MaxActive / 3; i++)
			{
				availableClientQueue.Enqueue(CreateClient());
			}
		}
		public CassandraHost CassandraHost
		{
			get
			{
				return cassandraHost;
			}
            set
            {
                cassandraHost = value;
            }
		}
		public string Name
		{
			get
			{
				return string.Format("<ConcurrentCassandraClientPoolByHost>:{0}", cassandraHost.Name);
			}
		}
		public int NumActive
		{
			get
			{
				return realActiveClientsCount.Value;
			}
		}
		public int NumBeforeExhausted
		{
			get
			{
				return cassandraHost.MaxActive - realActiveClientsCount.Value;
			}
		}
		public int NumBlockedThreads
		{
			get
			{
				return numBlocked.Value;
			}
		}
		virtual public int NumIdle
		{
			get
			{
				return availableClientQueue.Count;
			}
		}
		virtual public bool Exhausted
		{
			get
			{
				return NumBeforeExhausted == 0;
			}
		}
		public int MaxActive
		{
			get
			{
				return cassandraHost.MaxActive;
			}
            set
            {
                cassandraHost.MaxActive = value;
            }
		}
		public bool IsActive
		{
			get
			{
				return active.Value;
			}
            protected set
            {
                active.Value = value;
            }
		}
		public string StatusAsString
		{
			get
			{
				return string.Format("{0}; IsActive?: {1}; Active: {2}; Blocked: {3}; Idle: {4}; NumBeforeExhausted: {5}", Name,
                    IsActive, NumActive, NumBlockedThreads, NumIdle, NumBeforeExhausted);
			}
		}
		public virtual HClient BorrowClient()
		{
			if (!active.Value)
			{
				throw new HInactivePoolException("Attempt to borrow on in-active pool: " + Name);
			}
			
			HClient cassandraClient = null;
            availableClientQueue.TryDequeue(out cassandraClient);
			int currentActiveClients = activeClientsCount.Increment();
			
			try
			{
				if (cassandraClient != null)
				{
					if (cassandraClient.CassandraHost.MaxLastSuccessTimeMillis > 0 && cassandraClient.LastSuccessTime > 0 &&
                        (System.DateTime.Now.Ticks - 621355968000000000) / 10000 - cassandraClient.LastSuccessTime > cassandraClient.CassandraHost.MaxLastSuccessTimeMillis)
					{
						logger.Info(string.Format("Closing connection to {0} due to too long idle time of {1} ms", 
                            cassandraClient.CassandraHost.Host, (System.DateTime.Now.Ticks - 621355968000000000) / 10000 - cassandraClient.LastSuccessTime));
						cassandraClient.Close();
						cassandraClient = null;

                        monitor.IncrementCounter(CassandraClientMonitor.Counter.RENEWED_IDLE_CONNECTIONS);
					}
				}
				if (cassandraClient != null)
				{
					if (cassandraClient.CassandraHost.MaxConnectTimeMillis > 0 &&
                        (System.DateTime.Now.Ticks - 621355968000000000) / 10000 - cassandraClient.CreatedTime > cassandraClient.CassandraHost.MaxConnectTimeMillis)
					{
						logger.Info(string.Format("Closing connection to {0} due to too long existence time of {1} ms", 
                            cassandraClient.CassandraHost.Host, (System.DateTime.Now.Ticks - 621355968000000000) / 10000 - cassandraClient.CreatedTime));
						cassandraClient.Close();
						cassandraClient = null;
						
						monitor.IncrementCounter(CassandraClientMonitor.Counter.RENEWED_TOO_LONG_CONNECTIONS);
					}
				}
				if (cassandraClient == null)
				{
					
					if (currentActiveClients <= cassandraHost.MaxActive)
					{
						cassandraClient = CreateClient();
					}
					else
					{
						// We can't grow so let's wait for a connection to become available.
						cassandraClient = WaitForConnection();
					}
				}
				
				if (cassandraClient == null)
				{
					throw new HectorException("HConnectionManager returned a null client after aquisition - are we shutting down?");
				}
			}
			catch (System.SystemException e)
			{
				activeClientsCount.Decrement();
				throw e;
			}
			
			realActiveClientsCount.Increment();
			return cassandraClient;
		}
		private HClient WaitForConnection()
		{
			HClient cassandraClient = null;
			numBlocked.Increment();
			
			try
			{
				// wait and catch, creating a new one if the counts have changed. Infinite wait should just recurse.
				if (maxWaitTimeWhenExhausted == 0)
				{
					while (cassandraClient == null && active.Value)
					{
						try
						{
                            availableClientQueue.TryDequeue(out cassandraClient);
						}
						catch (System.Threading.ThreadInterruptedException ie)
						{
							logger.Error(string.Format("InterruptedException poll operation on retry forever {0}", ie));
							break;
						}
					}
				}
				else
				{
					try
					{
                        availableClientQueue.TryDequeue(out cassandraClient);
						if (cassandraClient == null)
						{
							throw new HPoolExhaustedException(string.Format("maxWaitTimeWhenExhausted exceeded for thread {0} on host {1}", 
                                System.Threading.Thread.CurrentThread.Name, cassandraHost.Name));
						}
					}
					catch (System.Threading.ThreadInterruptedException ie)
					{
						// monitor.incCounter(Counter.POOL_EXHAUSTED);
						logger.Error(string.Format("Cassandra client acquisition interrupted {)}", ie));
					}
				}
			}
			finally
			{
                numBlocked.Decrement();
			}
			
			return cassandraClient;
		}
		private HClient CreateClient()
		{
			return clientFactory.createClient(cassandraHost).Open();
		}
		public void Shutdown()
		{
            if (IsActive)
            {
                IsActive = false;
                logger.Info(string.Format("Shutdown triggered on {0}", Name));
                List<HClient> clients = new List<HClient>();
                foreach (HClient client in availableClientQueue)
                {
                    clients.Add(client);
                }
                ConcurrentQueueExtensions.Clear(availableClientQueue);
                if (clients.Count > 0)
                {
                    foreach (HClient hClient in clients)
                    {
                        try
                        {
                            hClient.Close();
                        }
                        catch (Exception e)
                        {
                            logger.Info(e.Message);
                        }
                    }
                }
                logger.Info(String.Format("Shutdown complete on {0}", Name));
            }
        }
		public virtual void ReleaseClient(HClient client)
		{
			if (cassandraHost.MaxActive == 0)
			{
				client.Close();
			}
            bool open = client.IsOpen();
			if (open)
			{
				if (active.Value)
				{
					AddClientToPoolGently(client);
				}
				else
				{
					logger.Info(string.Format("Open client {0} released to in-active pool for host {1}. Closing.", client, cassandraHost));
					client.Close();
				}
			}
			else
			{
				try
				{
					AddClientToPoolGently(CreateClient());
				}
				catch (HectorTransportException e)
				{
					// if unable to open client then don't add one back to the pool
					logger.Error(string.Format("Transport exception in re-opening client in release on {0} {1}", Name, e));
				}
			}

            realActiveClientsCount.Decrement();
            activeClientsCount.Decrement();
		}
		private void AddClientToPoolGently(HClient client)
		{
			try
			{
				availableClientQueue.Enqueue(client);
			}
			catch (System.SystemException ise)
			{
				logger.Warning("Capacity hit adding client back to queue. Closing extra. {0}" + ise.Message);
				client.Close();
			}
		}
	}
}